/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.handler;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import cn.ac.iie.di.ban.commons.RuntimeEnv;
import java.util.Map;
import cn.ac.iie.di.ban.metastore.Load2DBEntity;
import cn.ac.iie.di.ban.metastore.MetaStoreManager;
import cn.ac.iie.di.ban.tools.RocketMqHelper;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 *
 * @author sxl
 */
public class HttpSe2DBHandler {

    public volatile static boolean isStoped = true;
    private Load2DBEntity load2DBEntity = null;
    private RocketMqHelper rocketMqHelper;
    List<SE2DBWorker> se2DBWorkerSet = new ArrayList<SE2DBWorker>();
    private ExecutorService threadPool = Executors.newFixedThreadPool(20);
    private BlockingQueue<List<Map<String, Object>>> dataCache = new ArrayBlockingQueue<>((Integer) RuntimeEnv.getParam(RuntimeEnv.DATA_CACHE_SIZE));
    /*
     * 
     */
    private static Logger logger = Logger.getLogger(HttpSe2DBHandler.class.getName());

    private HttpSe2DBHandler(Load2DBEntity load2DBEntity) {
        this.load2DBEntity = load2DBEntity;
    }

    public static HttpSe2DBHandler getHttpSe2MqHandler(Load2DBEntity load2DBEntity) {
        HttpSe2DBHandler httpSe2DBHandler = new HttpSe2DBHandler(load2DBEntity);
        try {
            httpSe2DBHandler.appendSE2DWTask(httpSe2DBHandler.getThread());
        } catch (Exception ex) {
            logger.warn("init db worker for table unsuccessfully for " + ex.getMessage(), ex);
            httpSe2DBHandler = null;
        }
        return httpSe2DBHandler;
    }

    public void startReceiveMsg() throws Exception {
        rocketMqHelper = new RocketMqHelper(load2DBEntity.getMQName(), dataCache, load2DBEntity.getGroup());
        rocketMqHelper.registerMsgHandler();
        rocketMqHelper.starConsumer();
    }

    private void appendSE2DWTask(int pTaskNum) throws Exception {
        for (int i = 0; i < pTaskNum; i++) {
            SE2DBWorker se2DBWorker = null;
            try {
                se2DBWorker = SE2DWWorkerFactory.getSE2DBWroker(this, i);
                se2DBWorkerSet.add(se2DBWorker);
                threadPool.execute(se2DBWorker);
            } catch (Exception ex) {
                logger.error("execute serializing to db  is failed for " + ex.getMessage(), ex);
                throw ex;
            }
        }
    }

    private void stopSE2DBTask() {
        for (SE2DBWorker se2DBWorker : se2DBWorkerSet) {
            se2DBWorker.shutdown();
        }
    }

    public void stop() {
        logger.info("stop receive msg ...");
        rocketMqHelper.stopConsumer();
        int dataCacheSize = -1;
        while (true) {
            dataCacheSize = dataCache.size();
            if (dataCacheSize < 1) {
                logger.info("all data of has been cleared");
                break;
            } else {
                logger.info("waiting  data of to be cleared " + dataCacheSize);
                try {
                    Thread.sleep(3000);
                } catch (Exception ex) {
                }
            }
        }
        stopSE2DBTask();
        threadPool.shutdown();
        try {
            threadPool.awaitTermination(10, TimeUnit.SECONDS);
        } catch (Exception ex) {
            logger.warn("stop thread pool unsuccessfully for " + ex.getMessage(), ex);
        }
    }

//    public void disposeMessages(Map<String, Object> msg) {
//        try {
//            Map record = msg;
//            while (true) {
//                logger.debug("dataCache size:" + dataCache.size());
//                if (!dataCache.offer(record)) {
//                    if (System.currentTimeMillis() % 30 == 0) {
//                        logger.info("put to buffer fail,will retry ...");
//                    }
//                    try {
//                        Thread.sleep(200);
//                    } catch (Exception ex) {
//                    }
//                } else {
//                    logger.info("analysisSize " + dataCache.size() + " receMsgSize ");
//                    break;
//                }
//            }
//        } catch (Exception ex) {
//            logger.error("unknown exception " + ex.getMessage(), ex);
//        }
//    }
    public List<Map<String, Object>> getData() {
        try {
            return dataCache.poll();
        } catch (Exception ex) {
            return null;
        }
    }

    public Executor getExecutor() {
        return null;
    }

    public String getLoadType() {
        return load2DBEntity.getLoadType();
    }

    public Integer getThread() {
        return load2DBEntity.getThread();
    }

    public Integer getLoadNum() {
        return load2DBEntity.getLoadNum();
    }

    public Load2DBEntity getLoad2MBEntity() {
        return load2DBEntity;
    }

    public String getTable() {
        return load2DBEntity.getTable();
    }

    public static String getDriver() {
        return MetaStoreManager.getMetaStoreManger().getDb_driver();
    }

    public static String getUrl() {
        return MetaStoreManager.getMetaStoreManger().getDb_url();
    }

    public static String getUser() {
        return MetaStoreManager.getMetaStoreManger().getDb_user();
    }

    public static String getPwd() {
        return MetaStoreManager.getMetaStoreManger().getDb_pwd();
    }

    public static Integer getInitSize() {
        return MetaStoreManager.getMetaStoreManger().getDbsource_initSize();
    }

    public static Integer getMaxSize() {
        return MetaStoreManager.getMetaStoreManger().getDbsource_maxSize();
    }
}
